package com.changanw.jedis.redisqueue;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.Set;
import java.util.UUID;

/**
 * 延迟消息队列
 */
public class DelayMsgQueue {
    private Jedis jedis;
    private String queue;

    public DelayMsgQueue(Jedis jedis, String queue) {
        this.jedis = jedis;
        this.queue = queue;
    }

    /**
     * 消息入队
     * @param data  要发送的消息
     */
    public void queue(Object data){
        //构造一个消息对象
        JavaMsg msg = new JavaMsg();
        msg.setId(UUID.randomUUID().toString());
        msg.setData(data);
        //序列化
        ObjectMapper objectMapper = new ObjectMapper();
        try {
           String s = objectMapper.writeValueAsString(msg);
            System.out.println("msg publish"+ new Date());
           //消息发送，score为延迟时间
           jedis.zadd(queue,System.currentTimeMillis()+5000,s);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息消费
     */
    public void loop(){
        while (!Thread.interrupted()){
            //读取score在0到当前时间戳的消息
            Set<String> zrange = jedis.zrangeByScore(queue, 0, System.currentTimeMillis(), 0, 1);
            if (zrange.isEmpty()){
                //如果消息为空 休息500ms再读
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }
            //如果读取到了消息，则直接加载
            final String next = zrange.iterator().next();
            if(jedis.zrem(queue,next)>0){
                //抢到了，可以处理业务
                ObjectMapper o = new ObjectMapper();
                try {
                    final JavaMsg msg = o.readValue(next, JavaMsg.class);
                    System.out.println("receive msg:" + new Date() +">>>"+ msg);
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
